package top.ieei.demo.tftp;

import java.net.InetAddress;
import java.net.SocketException;
import java.util.Date;
import java.util.Hashtable;
import java.util.Stack;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

/**
 * 处理来自TftpSocket的TFTP读写请求<br>
 * @author xiecongshu
 */
public class TftpWorkPool
{
    private static final Log LOG = LogFactory.getLog(TftpWorkPool.class);
    
    /**
     * 空闲的TFTP请求处理线程堆栈
     */
    private Stack<RequestHandlerThread> idleWorkers;
    
    /**
     * TFTP工作池大小
     */
    private int size;
    
    /**
     * TFTP请求连接
     */
    private Hashtable<String, RequestHandler> connections;
    
    /**
     * <默认构造函数>
     */
    public TftpWorkPool(int size, Hashtable<String, RequestHandler> connections)
    {
        this.size = size;
        this.connections = connections;
        
        idleWorkers = new Stack<RequestHandlerThread>();
        RequestHandlerThread requestHandlerThread;
        for (int i = 0; i < size; i++)
        {
            String threadId = "TFTP RequestHandlerThread, worker id=" + i
                    + ",create time=" + (new Date()).getTime();
            try
            {
                requestHandlerThread = new RequestHandlerThread(threadId,
                        new RequestHandler());
                requestHandlerThread.start();
                idleWorkers.push(requestHandlerThread);
            }
            catch (SocketException e)
            {
                LOG.error("Could not create RequestHandlerThread" + i
                        + " for TftpWorkPool");
            }
        }
    }
    
    /**
     * 修改线程池大小，作用于减小工作池大小
     * @param newPoolSize void
     */
    public void resize(int newPoolSize)
    {
        if (newPoolSize < size)
        {
            for (int x = 0; x < (size - newPoolSize); x++)
            {
                if (idleWorkers.isEmpty())
                    break;
                
                RequestHandlerThread rht = (RequestHandlerThread) idleWorkers.pop();
                if (rht == null)
                {
                    break;
                }
                rht.goToDie();
            }
        }
        size = newPoolSize;
    }
    
    /**
     * 添加请求任务到工作池中
     * @param request
     * @param clientAddress
     * @param clientPort void
     */
    public synchronized void addToWorkPool(TftpRequest request,
            InetAddress clientAddress, int clientPort)
    {
        // 请求处理线程
        RequestHandlerThread rht = null;
        
        // 请求处理程序
        RequestHandler existConnection;
        
        // TFTP请求连接标记
        String key = clientAddress.getHostName() + ':' + clientPort;
        
        synchronized (connections)
        {
            existConnection = (RequestHandler) connections.get(key);
        }
        
        if (existConnection != null)
        {
            if (!existConnection.isIdle())
            {
                LOG.debug("It is already working for " + key);
                return;
            }
        }
        
        synchronized (idleWorkers)
        {
            if (idleWorkers.empty())
            {
                try
                {
                    Thread.sleep(2);
                }
                catch (InterruptedException e1)
                {
                    // TODO
                }
                
                // 所有工人都在工作，创建新的线程 处理这次请求
                String worker = "TFTP RequestHandlerThread, worker id=overload,create time="
                        + (new Date()).getTime();
                try
                {
                    LOG.warn("WARNING: Overload on TftpWorkPool! ReHash!");
                    rht = new RequestHandlerThread(worker, new RequestHandler());
                    rht.start();
                }
                catch (SocketException e)
                {
                    LOG.error("ERROR: Can not handle overload!");
                }
            }
            else
            {
                // 获取空闲的线程处理请求
                rht = (RequestHandlerThread) idleWorkers.pop();
            }
        }
        
        // 确保在添加到连接中，先唤醒线程
        synchronized (connections)
        {
            rht.wake(request, clientAddress, clientPort);
            connections.put(key, rht.getRequestHandler());
        }
    }
    
    private boolean push(RequestHandlerThread rht)
    {
        boolean stayAround = false;
        synchronized (idleWorkers)
        {
            if (idleWorkers.size() < size)
            {
                stayAround = true;
                idleWorkers.push(rht);
            }
        }
        return stayAround;
    }
    
    /**
     *  内部类，实际处理请求任务的线程
     */
    class RequestHandlerThread extends Thread
    {
        /**
         * TFTP请求处理器
         */
        private RequestHandler tftpRequestHandler;
        
        /**
         * TFTP请求包
         */
        private TftpRequest requestPacket;
        
        /**
         * 请求客户端地址
         */
        private InetAddress address;
        
        /**
         * 请求客户端端口
         */
        private int port;
        
        /**
         * 线程是否死掉
         */
        private boolean die = false;
        
        /**
         * <默认构造函数>
         */
        public RequestHandlerThread(String id, RequestHandler rh)
        {
            super(id);
            tftpRequestHandler = rh;
            requestPacket = null;
            address = null;
            port = 0;
        }
        
        /**
         * 获取TFTP请求处理程序
         * @return RequestHandler
         */
        public RequestHandler getRequestHandler()
        {
            return tftpRequestHandler;
        }
        
        /**
         * 
         * @param request
         * @param clientAddress
         * @param clientPort void
         */
        public synchronized void wake(TftpRequest request,
                InetAddress clientAddress, int clientPort)
        {
            requestPacket = request;
            address = clientAddress;
            port = clientPort;
            notify();
        }
        
        /**
         * 请求处理线程执行方法
         */
        public synchronized void run()
        {
            try
            {
                boolean stayAround = true;
                
                while (stayAround)
                {
                    if (die)
                    {
                        return;
                    }
                    if (requestPacket == null)
                    {
                        try
                        {
                            wait();
                        }
                        catch (InterruptedException e)
                        {
                            continue;
                        }
                    }
                    tftpRequestHandler.handleRequests(requestPacket,
                            address,
                            port);
                    synchronized (connections)
                    {
                        connections.remove(address.getHostAddress() + ':'
                                + port);
                    }
                    
                    requestPacket = null;
                    address = null;
                    port = 0;
                    stayAround = push(this);
                }
            }
            finally
            {
                tftpRequestHandler.stop();
            }
        }
        
        /**
         * 关闭线程
         *
         */
        public synchronized void goToDie()
        {
            die = true;
            // wake up this thread to die
            interrupt();
        }
    };
}
